package com.wzp.util.highconcurrency;


import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 测试结果：
 * [test1, test1, test1, test1]
 * Future方式耗时：173ms
 * <p>
 * [test1, test1, test1, test1]
 * CompletableFuture方式耗时：111ms
 * <p>
 * [test1, test1, test1, test1]
 * GuavaListenableFuture方式耗时：133ms
 *
 * @author wangzhengpeng
 */
public class HighConcurrencyTest {
    public static final int count = 50;
    private static int core = 50;
//    private static int core = Runtime.getRuntime().availableProcessors();
    private static ExecutorService executorFutureTest = new ThreadPoolExecutor(
            core,
            core,
            30,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<>(count * 2),
            new ThreadFactoryBuilder().setNameFormat("HighConcurrencyTest").build(),
            new ThreadPoolExecutor.AbortPolicy());

    private static ExecutorService executorHighConcurrencyTest = new ThreadPoolExecutor(
            core,
            core,
            30,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<>(count * 2),
            new ThreadFactoryBuilder().setNameFormat("HighConcurrencyTest").build(),
            new ThreadPoolExecutor.AbortPolicy());

    private static ExecutorService executorGuavaListenableFuture = new ThreadPoolExecutor(
            core,
            core,
            30,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<>(count * 2),
            new ThreadFactoryBuilder().setNameFormat("GuavaListenableFuture").build(),
            new ThreadPoolExecutor.AbortPolicy());

    private static ExecutorService executorCountLatchDown = new ThreadPoolExecutor(
            core,
            core,
            30,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<>(count * 2),
            new ThreadFactoryBuilder().setNameFormat("executorCountLatchDown").build(),
            new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("Runtime.getRuntime().availableProcessors() = " + core);
        futureMethod();
        completableFutureMethod();
        guavaListenableFutureMethod();
        countDownLatchMethod();
    }

    private static void futureMethod() {
        long start = System.currentTimeMillis();
        List<Future<String>> futures = Lists.newArrayList();
        for (int i = 0; i < count; i++) {
            futures.add(executorFutureTest.submit(HighConcurrencyTest::test1));
        }

        List<String> collect = futures.stream().map(future -> {
            try {
                return future.get();
            } catch (Exception e) {
                return null;
            }
        }).collect(Collectors.toList());
//        System.out.println(collect);
        System.out.println("Future方式耗时：" + (System.currentTimeMillis() - start) + "ms");
        System.out.println();
    }

    private static void completableFutureMethod() {
        long start = System.currentTimeMillis();
        ArrayList<CompletableFuture<String>> futures = Lists.newArrayListWithCapacity(4);
        for (int i = 0; i < count; i++) {
            futures.add(CompletableFuture.supplyAsync(HighConcurrencyTest::test1, executorHighConcurrencyTest));
        }

        CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[count]));

        CompletableFuture<List<String>> listCompletableFuture = allOf.thenApplyAsync(value ->
                futures.stream().map(CompletableFuture::join).collect(Collectors.toList()), executorHighConcurrencyTest);

        List<String> join = listCompletableFuture.join();
        System.out.println("CompletableFuture方式耗时：" + (System.currentTimeMillis() - start) + "ms");
        System.out.println();
    }

    private static void guavaListenableFutureMethod() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executorGuavaListenableFuture);
        ListenableFuture[] futures = new ListenableFuture[count];
        for (int i = 0; i < count; i++) {
            futures[i] = listeningExecutorService.submit(HighConcurrencyTest::test1);
        }
        ListenableFuture<List<String>> listListenableFuture = Futures.allAsList(futures);

        List<String> join = listListenableFuture.get();
        System.out.println("GuavaListenableFuture方式耗时：" + (System.currentTimeMillis() - start) + "ms");
        System.out.println();
    }

    private static void countDownLatchMethod() {
        long start = System.currentTimeMillis();
        final CountDownLatch latch = new CountDownLatch(count);
        for (int i = 0; i < count; i++) {
            executorCountLatchDown.submit(new Runnable() {
                @Override
                public void run() {
                    test1();
                    latch.countDown();
                }
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("CountDownLatch方式耗时：" + (System.currentTimeMillis() - start) + "ms");
        System.out.println();
    }


    private static String test1() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "test1";
    }
}
